package com.tech

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.row_number
import org.apache.spark.sql.{Dataset, Row, SparkSession}
import com.tech.common.KafkaUtil
import com.tech.config.ApplicationConfig
import org.apache.spark.sql.SparkSession

import java.time.LocalDate

object StaticGroupMonitor4D365 {

  def main(args: Array[String]): Unit = {
    println(s"创建 StaticGroupMonitor4D365 spark ${ApplicationConfig.PROJECT}")
    while (true) {
      foreachStaticGroup()
    }

  }

  def foreachStaticGroup(): Unit = {

    Logger.getLogger("org").setLevel(Level.WARN)

    println("创建spark")
    val spark = SparkSession
      .builder
      .appName("DynamicGroupMonitor")
      .config("spark.debug.maxToStringFields", "5000")
      //.config("spark.sql.session.timeZone", ApplicationConfig.TIMEZONE)
      //      .config("spark.sql.crossJoin.enabled","true")
      .config("spark.executor.cores", "1")
      .config("spark.executor.memory", "10g")
      .config("spark.local.dir", ApplicationConfig.SPARK_LOCAL_DIR)
      .master("local[*]")
      .getOrCreate
    println("读取sql")


    val df_tsr_marketing_activity = spark.read
      .format("org.apache.kudu.spark.kudu")
      .option("kudu.master", ApplicationConfig.KUDU_MASTER)
      .option("kudu.table", "tsr_marketing_activity")
      .load()

    df_tsr_marketing_activity.createOrReplaceTempView("tsr_marketing_activity")

    val df_tsr_marketing_process_customer = spark.read
      .format("org.apache.kudu.spark.kudu")
      .option("kudu.master", ApplicationConfig.KUDU_MASTER)
      .option("kudu.table", "tsr_marketing_process_customer")
      .load()

    df_tsr_marketing_process_customer.createOrReplaceTempView("tsr_marketing_process_customer")

    val now = LocalDate.now()

    //group_id activity_id process_id
    val df_group_activity_process = spark.sql(
      s"""
         |select
         |	mpc.customer_group_id group_id,
         |	ma.id activity_id,
         |	mpc.process_id process_id
         |from
         |	tsr_marketing_activity ma
         |left join
         |	tsr_marketing_process_customer mpc
         |on
         |	ma.id = mpc.activity_id
         |where
         |	ma.status = '100000002'
         |and
         |	current_timestamp() between ma.begin_time and ma.end_time
         |and
         |	mpc.customer_group_id is not null
         |and
         |	mpc.group_type = 100000000
         |and
         | (ma.activity_type = 0
         |   or
         |  (ma.activity_type = 1 and ma.period_unit = 100000000 and ma.period_value = ${now.getDayOfYear})
         |   or
         |  (ma.activity_type = 1 and ma.period_unit = 100000001 and ma.period_value = ${now.getDayOfMonth})
         |   or
         |  (ma.activity_type = 1 and ma.period_unit = 100000002 and ma.period_value = ${now.getDayOfWeek.getValue})
         |   or
         |  (ma.activity_type = 1 and ma.period_unit = 100000003)
         |  )
         |""".stripMargin)
    df_group_activity_process.show(100, false)

    val df_customer = spark.read
      .format("org.apache.kudu.spark.kudu")
      .option("kudu.master", ApplicationConfig.KUDU_MASTER)
      .option("kudu.table", "customer")
      .load()
      .filter(s"project = '${ApplicationConfig.PROJECT}'")
      .selectExpr("customer_id", "member_id")

    val df_customer_group_detail = spark.read
      .format("org.apache.kudu.spark.kudu")
      .option("kudu.master", ApplicationConfig.KUDU_MASTER)
      .option("kudu.table", "customer_group_detail")
      .load()

    df_customer_group_detail.createOrReplaceTempView("customer_group_detail")

    val df_customer_activity = spark.read
      .format("org.apache.kudu.spark.kudu")
      .option("kudu.master", ApplicationConfig.KUDU_MASTER)
      .option("kudu.table", "customer_activity")
      .load()
      .selectExpr("customer_id", "activity_id")

    val df_ma_group_record = spark.read
      .format("org.apache.kudu.spark.kudu")
      .option("kudu.master", ApplicationConfig.KUDU_MASTER)
      .option("kudu.table", "ma_group_record")
      .load()
      .selectExpr("activity_id", "process_id")

    //获取当前活跃活动包含的群组
    val df_group = df_group_activity_process.join(df_ma_group_record, Seq("activity_id", "process_id"), "left_anti").select("group_id","activity_id", "process_id").distinct()

    val rows = df_group.collect()

    val iterator = rows.iterator

    while (iterator.hasNext) {

      val row = iterator.next()
      val group_id = row.getString(0)
      val activity_id = row.getString(1)
      val process_id = row.getString(2)
      println(group_id)
      println("GroupStartTime:" + System.currentTimeMillis())



      val df_group_customer = df_customer_group_detail
        .filter(s"group_id = '$group_id'")
        .distinct()
        .selectExpr(s"'$group_id' as group_id", "cust_id as customer_id")
      //          .withColumn("id",row_number().over(Window.partitionBy().orderBy("customer_id")))
      //          .filter(s"id%16 = $i")

      println("FinalStartTime:" + System.currentTimeMillis())

      val df_final = df_group_customer
        .join(df_customer, Seq("customer_id"), "inner")
        .select("group_id", "customer_id", "member_id")
        .join(df_group_activity_process, Seq("group_id"), "inner")
        .join(df_customer_activity, Seq("customer_id", "activity_id"), "left_anti")
        .selectExpr("customer_id", "activity_id", "process_id", "member_id")
        .distinct()


      println("FinalEndTime:" + System.currentTimeMillis())

//      df_final
//        .selectExpr(s"'{\"project\":\"'" + s"'${GlobalValue.project}'" + "'\",\"activity_id\":\"'" + "activity_id" + "'\",\"process_id\":\"'" + "process_id" + "'\",\"customer_id\":\"'" + "customer_id" + "'\",\"member_id\":\"'" + "member_id" + "'\"}'")
//      .write
//        .format("kafka")
//        .option("kafka.bootstrap.servers", GlobalValue.broker)
//        .option("kafka.security.protocol", "SSL")
//        .option("topic",GlobalValue.dynamicCustomerTopic)
//        .save()

      df_final
        .foreach(
          i => {
            KafkaUtil.send2Kafka(ApplicationConfig.DYNAMIC_CUSTOMER_TOPIC, ApplicationConfig.PROJECT, i.getString(1), i.getString(2), i.getString(0), i.getString(3))
          }
        )

      println("GroupEndTime:" + System.currentTimeMillis())
    }

    df_group_activity_process
      .selectExpr("activity_id", "process_id", "current_timestamp() as create_time")
      .write
      .format("org.apache.kudu.spark.kudu")
      .option("kudu.master", ApplicationConfig.KUDU_MASTER)
      .option("kudu.table", "ma_group_record")
      .mode("append")
      .save()


    println("执行完毕，休眠5秒")

    Thread.sleep(5000)

    spark.close()
  }


}
